#include "TinyTCPServer.h"


#define TCPHBTIMEOUT             210
#define SENDMSG_WAIT_TIMEOUT     1

#define INITRECVTHREADCOUNT      16
#define INITSENDTHREADCOUNT      4

#define MAXRECVTHREAD            100
#define MAXSENDTHREAD            50

#define BINDTRYCOUNT             2

namespace TinyCC{
	namespace TCP{

		CTinyTCPServer::CTinyTCPServer(void)
			:m_pRecvThread( NULL )
		{
			WSADATA wsaData;
			if (WSAStartup(0x202,&wsaData) == SOCKET_ERROR) 
			{
				char buf[64];
				sprintf_s( buf, 64, "WSAStartup failed with error %d.",WSAGetLastError() );
				//g_Debug.debugPrint( buf );
			}
		}

		CTinyTCPServer::~CTinyTCPServer(void)
		{
			if( m_pRecvThread != NULL )
			{
				delete m_pRecvThread;
				m_pRecvThread = NULL;
			}
			//end wsa needed?
			WSACleanup();
		}

		void CTinyTCPServer::clear()
		{
			if( m_pRecvThread != NULL )
			{
				delete m_pRecvThread;
				m_pRecvThread = NULL;
			}
		}

		void CTinyTCPServer::init()
		{
			Lock lock(this);
			m_pRecvThreadPool=new CTinyThreadPool(MAXRECVTHREAD);
			for(int i=0;i<INITRECVTHREADCOUNT;)
			{
				CReceiveThreadInfo* pThreadInfo=new CReceiveThreadInfo();
				CTinyThread* pNewThread = m_pRecvThreadPool->createNewThread(this,&CTinyTCPServer::_recvThread,(void*)this);
				if( pNewThread != NULL )
				{
					i++;
					pThreadInfo->setThread(pNewThread);
					m_mapRecvThreadInfo.insert(pair<DWORD,CReceiveThreadInfo*>(pThreadInfo->getThreadId(),pThreadInfo));
				}
				else
				{
					delete pThreadInfo;
					pThreadInfo=NULL;
					delete pNewThread;
					pNewThread=NULL;
					break;
				}
			}

			m_pSendThreadPool=new CTinyThreadPool(MAXSENDTHREAD);
			for(int i=0;i<INITSENDTHREADCOUNT;++i)
			{
				CSendThreadInfo* pThreadInfo=new CSendThreadInfo();
				CTinyThread* pNewThread = m_pSendThreadPool->createNewThread(this ,&CTinyTCPServer::_sendThread,(void*)this);
				if( pNewThread != NULL ){
					i++;
					pThreadInfo->setThisThreadHandle(pNewThread->getThreadHandle());
					pThreadInfo->setThreadId(pNewThread->getThreadId());
					m_mapSendThreadInfo.insert(pair<DWORD,CSendThreadInfo*>(pThreadInfo->getThreadId(),pThreadInfo));
				}
				else
				{
					delete pThreadInfo;
					break;
				}
			}
		}

		bool CTinyTCPServer::startRecv( const char* ip, int port, HANDLE recEvent )
		{
			sprintf_s(m_serverip,32,"%s",ip);
			m_serverport = port;
			//listen special ip&port
			char *pszInterface = NULL;
			long Sum = 0;

			m_serversock = socket(PF_INET, SOCK_STREAM,0); // TCP socket
			if (m_serversock == INVALID_SOCKET){
				//fprintf(stderr,"socket() failed with error %d\n",WSAGetLastError());
				return false;
			}

			SOCKADDR_IN	SockAddr;
			memset(&SockAddr,0, sizeof(SockAddr));
			SockAddr.sin_family = AF_INET;

			BYTE IPAddr[4];

			getIPAsByte(&IPAddr[0]);

			memcpy(&(SockAddr.sin_addr),IPAddr,4);
			SockAddr.sin_port   = htons( m_serverport );

			//set addr reuse on socket layer option
			int on = 1;
			setsockopt(m_serversock,SOL_SOCKET,SO_REUSEADDR,(char*)&on,sizeof(on));
			// Bind to the address and port
			for(int trycount=0; trycount<BINDTRYCOUNT;trycount++ )
			{
				if(bind(m_serversock, (sockaddr*)&SockAddr, sizeof(SockAddr)) != 0 )
				{
					int error = WSAGetLastError();

					Sleep( 10000 );	
				}
				else
				{
					break;
				}
			}//while

			if(listen(m_serversock,SOMAXCONN) != 0){
				//sprintf_s(buf,128,"start server on %s:%d err\n",ps->m_serverip,ps->m_serverport);
				return false;
			}
			//end listen
			if( m_pRecvThread == NULL )
			{
				m_pRecvThreadPool->setRecHandle( recEvent );
				m_pRecvThread = new CTinyThread( this, &CTinyTCPServer::acceptThread, (void*)this );
				return true;
			}

			return true;
		}

		bool CTinyTCPServer::stopRecv()
		{
			//maybe exception,need repair.
			if(m_pRecvThreadPool!=NULL){
				m_pRecvThreadPool=NULL;
			}
			if( m_pRecvThread != NULL )
			{
				delete m_pRecvThread;
				m_pRecvThread = NULL;
			}
			closesocket(m_serversock);

			return true;
		}


		bool CTinyTCPServer::getIPAsByte(BYTE* IPByte)
		{
			const char* pszAddr = m_serverip;
			long Flag,Len;
			char IPSection[8];
			//BYTE IPAddr[4];

			//tcpserver->getIPAsByte(&IPAddr);

			Flag = strcspn(pszAddr,".");
			Len = strlen(pszAddr);
			if(Flag <= 0 || Flag >= Len)
				return false;
			memset(IPSection,0,8);
			memcpy(IPSection,pszAddr,Flag);
			IPByte[0] = atoi(IPSection);
			pszAddr += Flag + 1;

			Flag = strcspn(pszAddr,".");
			Len = strlen(pszAddr);
			if(Flag <= 0 || Flag >= Len)
				return false;
			memset(IPSection,0,8);
			memcpy(IPSection,pszAddr,Flag);
			IPByte[1] = atoi(IPSection);
			pszAddr += Flag + 1;

			Flag = strcspn(pszAddr,".");
			Len = strlen(pszAddr);
			if(Flag <= 0 || Flag >= Len)
				return false;
			memset(IPSection,0,8);
			memcpy(IPSection,pszAddr,Flag);
			IPByte[2] = atoi(IPSection);
			pszAddr += Flag + 1;

			Len = strlen(pszAddr);
			memset(IPSection,0,8);
			memcpy(IPSection,pszAddr,Len);
			IPByte[3] = atoi(IPSection);

			return true;
		}
		void CTinyTCPServer::acceptThread( void* pParam)
		{
			CTinyTCPServer* tcpserver = (CTinyTCPServer*)pParam;

			DWORD dwThreadId = 0;

			dwThreadId = GetCurrentThreadId();


			while(1)
			{
				SOCKADDR_IN	fromSockAddr;
				int socklen = sizeof(sockaddr);
				SOCKET socketid = accept(tcpserver->m_serversock,(sockaddr *)&fromSockAddr,&socklen);
				std::string strClientIP;
				char clientip[32];
				sprintf_s(clientip,32,"%d.%d.%d.%d",
					fromSockAddr.sin_addr.S_un.S_un_b.s_b1,
					fromSockAddr.sin_addr.S_un.S_un_b.s_b2,
					fromSockAddr.sin_addr.S_un.S_un_b.s_b3,
					fromSockAddr.sin_addr.S_un.S_un_b.s_b4);
				strClientIP = clientip;


				if(socketid == SOCKET_ERROR ){
					//WSAGetLastError
					continue;
				}
				tcpserver->addTCPClient(socketid,strClientIP.c_str());
			}

			return;
		}
		DWORD CTinyTCPServer::addTCPClient(SOCKET hAccept,const char* remoteIP)
		{
			DWORD dwThreadID;
			Lock lock(*this);

			//Find a thread not full in the thread pool.

			std::map<DWORD,CReceiveThreadInfo*>::iterator itEnumThreadInfo=m_mapRecvThreadInfo.begin();
			while(itEnumThreadInfo != m_mapRecvThreadInfo.end() )
			{
				if(itEnumThreadInfo->second->getConnectionCount()<MAXCONNECTIONPERTHREAD)
				{
					itEnumThreadInfo->second->addSocketId(hAccept);
					itEnumThreadInfo->second->setConnectedEvent();
					return itEnumThreadInfo->second->getThreadId();
				}
				itEnumThreadInfo++;
			}//while

			//add new thread to process more connection
			/*if(itEnumThreadInfo==m_mapRecvThreadInfo.end())
			{
			/*if(m_mapRecvThreadInfo.size()>=m_pRecvThreadPool->getMaxThreadCount()){
			//			g_Debug.debugPrint("Expired the max receive thread count.");
			return 0;
			}//if

			//HANDLE hNewThreadHandle;
			//DWORD dwNewThreadId;
			CReceiveThreadInfo* pThreadInfo=new CReceiveThreadInfo();
			CTinyThread* pNewThread = m_pRecvThreadPool->createNewThread(this, &CTinyTCPServer::_recvThread, (void*)this );
			if( pNewThread != NULL ){
			pThreadInfo->setThread(pNewThread);
			pThreadInfo->setSocketId(hAccept);
			pThreadInfo->setConnectedEvent();
			pThreadInfo->setIsConnected(true);
			pThreadInfo->setRemoteIP(remoteIP);
			m_mapRecvThreadInfo.insert(pair<DWORD,CReceiveThreadInfo*>(pThreadInfo->getThreadId(),pThreadInfo));
			return pThreadInfo->getThreadId();
			}
			}*/
			return 0;
		}

		bool CTinyTCPServer::sendData(SOCKET socketID, void *data, long datalen)
		{
			m_SendMsgBuff.push(socketID,(char*)data,datalen);
			sendInfo();
			return true;
		}


		void CTinyTCPServer::_sendThread(void* pParam)
		{
			//Lock lock(this);
			Sleep(50);

			CTinyTCPServer *pTCPServer=(CTinyTCPServer*)pParam;
			if(pTCPServer==NULL)
				return;

			DWORD dwThreadId = 0;
			dwThreadId = GetCurrentThreadId();
			CSendThreadInfo* pThreadInfo = NULL;
			{

				pThreadInfo=m_mapSendThreadInfo[dwThreadId];

			}

			if(pThreadInfo==NULL)
				return;

			DWORD dwIndex=0;
			int nTemp=0;
			if(pThreadInfo->getDeadEvent()==NULL)
				nTemp=2;
			else
				nTemp=3;
			HANDLE* pEvent=new HANDLE[nTemp];
			pEvent[0]=pTCPServer->m_pSendThreadPool->getAllDeadEvent();
			pEvent[1]=pThreadInfo->getWorkEvent();
			if(nTemp==3)
				pEvent[2]=pThreadInfo->getDeadEvent();
			pTCPServer->m_pSendThreadPool->increaseThread();

			UINT nCount=0;

			try{
				while(true)
				{
					dwIndex=WaitForMultipleObjects(nTemp,pEvent,false,50);
					if(dwIndex==WSA_WAIT_FAILED){
						Sleep(300);
						continue;
					}

					DWORD dwTemp=dwIndex-WSA_WAIT_EVENT_0;
					switch(dwTemp){
		case 0:	
		case 2:	
			goto CLEAR;
			break;
		case 1:
		case WAIT_TIMEOUT:
			break;
		default:	
			continue;
			break;
					}
					TinyCC::msg::CTinyTCPRawMsg* pRawMsg = pTCPServer->m_SendMsgBuff.pop();
					if(pRawMsg==NULL)
					{
						pThreadInfo->setIsWork(false);
						pThreadInfo->resetWorkEvent();
						continue;
					}
					if(pThreadInfo->getIsWork()==false)
					{
						//pTCPServer->m_pSendThreadPool->increaseThread();

						pThreadInfo->setIsWork(true);
					}

					TIMEVAL tv={SENDMSG_WAIT_TIMEOUT,0};

					FD_SET fdWrite={1,pRawMsg->m_socketID};
					FD_SET fdException={1,pRawMsg->m_socketID};
					int nSelectResult=select(0,NULL,&fdWrite,&fdException,&tv);


					if(nSelectResult==0){
						delete pRawMsg;
						continue;	
					}
					else if(nSelectResult==	SOCKET_ERROR){
						delete pRawMsg;
						Sleep(300);
						continue;
					}
					else
					{
						if(FD_ISSET(pRawMsg->m_socketID,&fdException)){
							closesocket(pRawMsg->m_socketID);
							pThreadInfo->setIsWork(false);
							pThreadInfo->resetWorkEvent();
							pTCPServer->m_pSendThreadPool->decreaseThread();

							delete pRawMsg;

						}
						else if(FD_ISSET(pRawMsg->m_socketID,&fdWrite))
						{
							int wRetVal;

							if( pRawMsg->m_pData != NULL && pRawMsg->m_lDataLength > 0)
							{
								wRetVal = send(pRawMsg->m_socketID,(const char*)pRawMsg->m_pData,pRawMsg->m_lDataLength,0);
								if (wRetVal<=0) 
								{
									pTCPServer->m_pSendThreadPool->decreaseThread();
									delete pRawMsg;
								}
								else
								{
									delete pRawMsg;
									nCount++;
								}
							}

							pThreadInfo->setIsWork(false);
							pThreadInfo->resetWorkEvent();
						}
					}

				}//while(true)
			}
			catch(...){
				//g_Debug.debugPrint("Send thread catch exceptin. ");
			}

CLEAR:
			if(pEvent!=NULL){
				delete[] pEvent;
				pEvent=NULL;
			}
			if(pThreadInfo->getIsWork()){
				//closesocket(pThreadInfo->getSocketId());
				pThreadInfo->setIsWork(false);
				pThreadInfo->resetWorkEvent();

				pTCPServer->m_pSendThreadPool->decreaseThread();

			}
			if(pThreadInfo->getThisThreadHandle()!=NULL)
				CloseHandle(pThreadInfo->getThisThreadHandle());
			if(pThreadInfo->getWorkEvent()!=NULL)
				CloseHandle(pThreadInfo->getWorkEvent());
			if(pThreadInfo->getDeadEvent()!=NULL)
				CloseHandle(pThreadInfo->getDeadEvent());
			{
				//Lock lock(this);
				map<DWORD,CSendThreadInfo*>::iterator itFindThreadInfo=pTCPServer->m_mapSendThreadInfo.find(dwThreadId);
				if(itFindThreadInfo!=pTCPServer->m_mapSendThreadInfo.end()){
					delete itFindThreadInfo->second;
					pTCPServer->m_mapSendThreadInfo.erase(itFindThreadInfo);
				}//if
			}

			pTCPServer->m_pSendThreadPool->decreaseThread();

			return;
		}

		bool CTinyTCPServer::getNextMsg(TinyCC::msg::CTinyMsg * msg,SOCKET &socketID)
		{
			TinyCC::msg::CTinyTCPRawMsg * tcprawmsg = m_RecvMsgBuff.pop();
			
			if(tcprawmsg !=NULL)
			{
				socketID = tcprawmsg->m_socketID;
				if(m_tcpMsg.unPackage( *tcprawmsg,msg))
				{
					msg->setSocketID(socketID);
					return true;
				}
			}
			return false;
		}
		void CTinyTCPServer::_recvThread(void* pParam)
		{
			Sleep(300);

			CTinyTCPServer			*pTCPServer=(CTinyTCPServer*)pParam;
			CReceiveThreadInfo		*pThreadInfo= NULL;
			DWORD					dwThreadId = 0;
			char					strLog[512];
			char					Buffer[TINYSOCKETTCPSIZE];
			DWORD					dwIndex=0;
			int						nEventNum=2;
			SOCKET					currentSockFD=-1;
			int						nSelectResult = 0;
			TIMEVAL					tv={RECEIVEMSG_WAIT_TIMEOUT,0};

			if(pTCPServer==NULL)
				return ;

			dwThreadId = GetCurrentThreadId();
			pThreadInfo=pTCPServer->m_mapRecvThreadInfo[dwThreadId];

			if(pThreadInfo==NULL)
				return ;

			HANDLE* pEvent=new HANDLE[nEventNum];
			pEvent[0]=pTCPServer->m_pRecvThreadPool->getAllDeadEvent();
			pEvent[1]=pThreadInfo->getConnectedEvent();
			pTCPServer->m_pRecvThreadPool->increaseThread();

			while(true)
			{
				dwIndex=WaitForMultipleObjects(nEventNum,pEvent,false,INFINITE);
				switch(dwIndex-WSA_WAIT_EVENT_0)
				{
				case WSA_WAIT_FAILED:		//timeout or error,retry.
					{
						Sleep(300);
						continue;
					}
					break;
				case 0:				//end thread
					{
						if(pEvent!=NULL){
							delete[] pEvent;
							pEvent=NULL;
						}
						//close sockets
						for(int i =0;i<MAXCONNECTIONPERTHREAD;++i)
							closesocket(pThreadInfo->m_selectSockets[i]);
						pThreadInfo->resetConnectedEvent();
						//release handle
						if(pThreadInfo->getThisThreadHandle()!=NULL)
							CloseHandle(pThreadInfo->getThisThreadHandle());
						if(pThreadInfo->getConnectedEvent()!=NULL)
							CloseHandle(pThreadInfo->getConnectedEvent());
						//free threadinfo
						Lock lock(this);
						map<DWORD,CReceiveThreadInfo*>::iterator itFindThreadInfo=pTCPServer->m_mapRecvThreadInfo.find(dwThreadId);
						if(itFindThreadInfo!=pTCPServer->m_mapRecvThreadInfo.end())
						{
							delete itFindThreadInfo->second;
							pTCPServer->m_mapRecvThreadInfo.erase(itFindThreadInfo);
						}
						//decrease thread count
						pTCPServer->m_pRecvThreadPool->decreaseThread();
						return;
					}
					break;
				case 1:			//event be set,select and process it
					{
						//fill m_fdRead for get state
						FD_ZERO(&(pThreadInfo->m_fdRead));
						for(int i=0;i<MAXCONNECTIONPERTHREAD;++i)
						{
							if(pThreadInfo->m_selectSockets[i]!=0)
								FD_SET(pThreadInfo->m_selectSockets[i], &(pThreadInfo->m_fdRead));
						}
						nSelectResult=select(0,&(pThreadInfo->m_fdRead),NULL,&(pThreadInfo->m_fdException),&tv);

						//select timeout, continue;
						if(nSelectResult == 0)
						{
							SetEvent( pThreadInfo->getConnectedEvent() );
							continue;
						}
						//select error, sleep and continue;
						if(nSelectResult == SOCKET_ERROR)
						{
							//no event happen or error,retry....
							SetEvent( pThreadInfo->getConnectedEvent() );
							Sleep(300);
							continue;	
						}
						else
						{
							SetEvent( pThreadInfo->getConnectedEvent() );
							//process event per socket
							for (int i = 0; i <= MAXCONNECTIONPERTHREAD; i++)
							{
								if ( (currentSockFD = pThreadInfo->m_selectSockets[i]) <= 0)
									continue;
								//count tcphb timeout
								if(pThreadInfo->m_TCPDuration[i]<=0)
								{
									pThreadInfo->removeSocketId(currentSockFD);
									closesocket(currentSockFD);
									if(pThreadInfo->getConnectionCount() <= 0)
										pThreadInfo->resetConnectedEvent();
									continue;
								}
								else
									pThreadInfo->m_TCPDuration[i]-=3;
								//end count*/

								if(FD_ISSET(currentSockFD,&(pThreadInfo->m_fdException)))
								{
									//socket exception,close connection
									closesocket(currentSockFD);
									pThreadInfo->removeSocketId(currentSockFD);
									if(pThreadInfo->getConnectionCount() <= 0)
										pThreadInfo->resetConnectedEvent();
									continue;
								}
								//read
								if(FD_ISSET(currentSockFD,&(pThreadInfo->m_fdRead)))
								{
									int wRetVal;
									memset(Buffer,0,TINYSOCKETTCPSIZE);
									wRetVal = recv(currentSockFD,Buffer,TINYSOCKETTCPSIZE,0);

									if (wRetVal == SOCKET_ERROR||wRetVal == 0) 
									{
										//connection error or closed
										pThreadInfo->removeSocketId(currentSockFD);
										closesocket(currentSockFD);
										if(pThreadInfo->getConnectionCount() <= 0)
											pThreadInfo->resetConnectedEvent();
										continue;
									}
									else
									{
										pTCPServer->processReceivedMsg(currentSockFD,Buffer,wRetVal);
										pThreadInfo->m_TCPDuration[i]=TCPHBTIMEOUT;
									}
								}//end read if
							}//end for
						}//end else
					}
					break;
				default:
					continue;
					break;
				}//end switch

			}//while(true)
		return;
	}


	bool CTinyTCPServer::sendInfo()
	{
		list<CSendThreadInfo*> lstThreadInfo;
		Lock lock(this);
		map<DWORD,CSendThreadInfo*>::iterator itEnumThreadInfo=m_mapSendThreadInfo.begin();
		while(itEnumThreadInfo!=m_mapSendThreadInfo.end())
		{
			CSendThreadInfo* pThreadInfo = itEnumThreadInfo->second;
			if( pThreadInfo != NULL )
				pThreadInfo->setWorkEvent();
			itEnumThreadInfo++;
		}//while

		return true;
	}

}//end namespace
}
int TinyCC::TCP::CTinyTCPServer::processReceivedMsg(SOCKET sockedfd,char* msg,long length)
{
	string msgstr = "";
	TinyCC::msg::CTCPHead	m_TCPHead; 
	long lProcessDataLen = 0;
	msgstr.append(msg,length);

	char* pDataCursor = (char*)msgstr.c_str();
	memcpy( (void*)(m_TCPHead.MagicKey), msgstr.c_str(), 4 );
	if( memcmp( m_TCPHead.MagicKey, TinyCC::msg::g_AtomMagicKey, 4 ) == 0 )
	{
		if(!TinyCC::msg::CTCPAtomMsg<TinyCC::msg::CTCPHBHead>::ParseMsgHead(m_TCPHead, msgstr.c_str(),msgstr.size())){
			return -1 ;
		}

		//if Atom HB Msg
		if(m_TCPHead.m_wMsgFunc==MSGTYPE_HB)
		{
			//send retrun msg if Client to Server
			if(memcmp(m_TCPHead.m_DestINS,"Server",8)==0)
			{
				TinyCC::msg::CTCPAtomMsg<TinyCC::msg::CTCPHBHead> m_HBMsg;
				m_HBMsg.setDestIns("Client");
				long length=0;
				const char * data =m_HBMsg.serialize(length);
				sendData(sockedfd,(void*)data,length);
			}
			msgstr.erase(0,msgstr.size()); 		//clear Msg string
			return 1;
		}
		else
		{
			//other type AtomMsg
		}
	}//end if
	else        //normalMsg
	{
		lProcessDataLen = m_RecvMsgBuff.push(sockedfd,msgstr.c_str(),msgstr.size());
		if(lProcessDataLen > 0)
		{
			if(lProcessDataLen >= msgstr.size())
			{
				msgstr="";
			}
			else
			{
				msgstr.erase(0, lProcessDataLen);
			}

			if( m_pRecvThreadPool->getRecHandle() != NULL )
				SetEvent( m_pRecvThreadPool->getRecHandle() );
		}
	}
	return 0;
}

bool TinyCC::TCP::CTinyTCPServer::isConnected(SOCKET socketID)
{
	std::map<DWORD,CReceiveThreadInfo*>::iterator itEnumThreadInfo=m_mapRecvThreadInfo.begin();
	while(itEnumThreadInfo != m_mapRecvThreadInfo.end() )
	{
		for(int i =0;i<MAXCONNECTIONPERTHREAD;++i)
		{
			if(itEnumThreadInfo->second->m_selectSockets[i] == socketID)
				return true;
		}
		itEnumThreadInfo++;
	}//while
	return false;
}
